package com.tlb.springboot.starter.rocketmq.impl;

import com.aliyun.openservices.shade.com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.aliyun.openservices.shade.com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.aliyun.openservices.shade.com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.aliyun.openservices.shade.com.alibaba.rocketmq.common.message.MessageExt;
import com.tlb.springboot.starter.rocketmq.annotation.RocketMQListen;
import lombok.extern.slf4j.Slf4j;

import java.lang.reflect.Method;
import java.util.List;

/**
 * @author yuaneg 2018/8/30
 */
@Slf4j
public class DefaultMessageListenerConcurrently extends AbstractMessageListener implements MessageListenerConcurrently {


    public DefaultMessageListenerConcurrently(Object bean, Method method, RocketMQListen rocketMQListen) {
        super(bean, method, rocketMQListen);
    }

    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> messageExts, ConsumeConcurrentlyContext context) {
        for (MessageExt messageExt : messageExts) {
            try {
                invokeMethod(messageExt);
            } catch (Exception e) {
                context.setDelayLevelWhenNextConsume(rocketMQListen.delayLevel());
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
}